%% @doc EMQX RabbitMQ插件的主模块。
%% 该模块负责插件的加载、卸载、重载逻辑，以及处理MQTT消息并将其转发到RabbitMQ。
%% 对于Java开发者：可以将其看作插件的入口类或主服务类，管理插件的生命周期和核心业务逻辑。
-module(emqx_plugin_rabbitmq).

%% 引入EMQX日志头文件，用于记录日志，类似于Java中的slf4j或log4j。
-include_lib("emqx/include/logger.hrl").
%% 引入插件自定义的头文件，通常包含宏定义和记录定义。
%% ?PLUGIN_rabbitmq_TAB, ?PLUGIN_rabbitmq_RESOURCE_ID 等宏在此定义。
-include("emqx_plugin_rabbitmq.hrl").
%% 引入EMQX核心头文件。
-include_lib("emqx/include/emqx.hrl").
%% 引入EMQX钩子相关的头文件。
-include_lib("emqx/include/emqx_hooks.hrl").

%% @doc 导出插件生命周期管理函数。
%% 这些函数会被EMQX框架在插件加载、卸载、重载时调用。
%% 类似于Java中某些框架定义的生命周期方法 (e.g., Spring Bean的 @PostConstruct, @PreDestroy)。
-export([
    load/0,    %% 插件加载时调用
    unload/0,  %% 插件卸载时调用
    reload/0   %% 插件配置重载时调用
]).

%% @doc 导出消息处理回调函数。
%% 当特定事件发生时（如此处的消息发布），EMQX框架会调用这些函数。
%% 类似于Java中的事件监听器方法。
-export([
    on_message_publish/1 %% 当MQTT消息发布时调用
]).

%% @doc 定义事件处理模块的名称为一个宏。
%% ?evt_mod 将被替换为 emqx_plugin_rabbitmq_evt。
%% 类似于Java中的 private static final String EVT_MODULE_NAME = "...";
-define(evt_mod, emqx_plugin_rabbitmq_evt).

%% @doc 插件加载函数。
%% 当EMQX启动并加载此插件时调用此函数。
load() ->
    %% 创建一个ETS (Erlang Term Storage) 表，用于存储主题配置规则。
    %% ?PLUGIN_rabbitmq_TAB 是在 .hrl 文件中定义的宏，代表表的名称。
    %% named_table: 表有一个全局名称。
    %% public: 任何进程都可以读写。
    %% set: 表类型，键唯一。
    %% {keypos, 1}: 元组的第一个元素作为键。
    %% {read_concurrency, true}: 优化并发读。
    %% 类似于Java中创建一个全局的 ConcurrentHashMap 来存储配置。
    _ = ets:new(?PLUGIN_RABBITMQ_TAB, [named_table, public, set, {keypos, 1}, {read_concurrency, true}]),
    %% 调用load/1函数，传入读取到的配置。
    load(read_config()).

%% @doc 根据配置加载插件 (多子句函数，通过模式匹配选择执行哪个子句)。
%% Conf = #{connection := _, producer := _, topics := Topics}
%% 这个模式匹配确保传入的Conf是一个Map，并且包含 connection, producer, topics 三个键。
%% 类似于Java中的方法重载，但基于参数的结构和值进行匹配。
load(Conf = #{connection := _, producer := ProducerConf = #{persistent_messages := PersistentMessages}, topics := Topics}) ->
    %% 启动插件所需的资源 (例如rabbitmq生产者)。
    {ok, _} = start_resource(Conf),
    %% 存储生产者配置，特别是 persistent_messages
    persistent_term:put(?PRODUCER_CONFIG_KEY, ProducerConf),
    %% 解析配置文件中的主题规则，并存入ETS表。
    topic_parse(Topics),
    %% 注册钩子函数。当 'message.publish' 事件发生时，调用本模块的 on_message_publish/1 函数。
    %% ?HP_HIGHEST 表示钩子具有最高优先级。
    %% 类似于Java中注册一个事件监听器。
    hook('message.publish', {?MODULE, on_message_publish, []});
%% 如果传入的配置不符合上面的模式 (例如，配置项缺失)，则执行这个子句。
load(_) ->
    {error, "config_error"}. %% 返回错误，表示配置错误。

%% @doc 辅助函数：添加钩子。
%% HookPoint: 钩子点名称 (原子类型)，如 'message.publish'。
%% MFA: {Module, Function, Args}，指定回调函数。
%% 这是一个通用的钩子注册函数。
hook(HookPoint, MFA) ->
    emqx_hooks:add(HookPoint, MFA, _Property = ?HP_HIGHEST).

%% @doc 重新加载插件配置。
%% 当插件配置发生变化并需要热重载时调用。
reload() ->
    ?SLOG(info, #{msg => "reloading_rabbitmq_plugin_config"}), %% 记录日志
    %% 清除ETS表中现有的所有主题配置规则。
    ets:delete_all_objects(?PLUGIN_RABBITMQ_TAB),
    %% 重新读取配置文件。
    NewConf = read_config(),
    %% 调用reload/1函数，传入新的配置。
    reload(NewConf).

%% @doc 根据新配置重新加载插件 (多子句函数)。
reload(Conf = #{connection := _, producer := _, topics := Topics}) ->
    %% 尝试使用新配置重新启动资源。
    case start_resource(Conf) of
        {ok, _} -> %% 资源启动成功
            %% 解析新的主题配置。
            topic_parse(Topics),
            ?SLOG(info, #{msg => "rabbitmq_plugin_config_reloaded_successfully"}),
            ok; %% 返回ok表示重载成功
        {error, Reason} -> %% 资源启动失败
            ?SLOG(error, #{msg => "failed_to_restart_resource", reason => Reason}),
            {error, Reason} %% 返回错误
    end;
reload(_) -> %% 配置格式无效
    ?SLOG(error, #{msg => "invalid_configuration_format"}),
    {error, "config_error"}.

%% @doc 消息发布回调处理函数。
%% 当MQTT客户端发布一条消息到Broker时，此函数会被调用。
%% Message: #message{} 记录，包含了消息的详细信息 (如主题、载荷、来源等)。
%% 类似于Java中事件监听器的 onEvent(MessageEvent event) 方法。

%% 子句1: 如果消息主题以 "$SYS/" 开头 (系统消息)，则直接跳过，不处理。
%% <<"$SYS/", _/binary>> 是二进制模式匹配，匹配以 "$SYS/" 开头的二进制串。
on_message_publish(Message = #message{topic = <<"$SYS/", _/binary>>}) ->
    {ok, Message}; %% 直接返回原始消息，表示不对此消息做任何修改或额外处理

%% 子句2: 处理普通消息。
%% #message{from = From} 从消息记录中提取 'from' 字段 (发布客户端的ID)。
%% @doc 处理MQTT消息发布钩子。
%% 当有MQTT客户端发布消息时，EMQX框架会调用此函数。
%% Message: EMQX内部的消息记录 (#message{})。
%% 返回 {ok, Message} 表示消息继续传递给下一个钩子或正常处理。
%% 返回 ignore 表示忽略此消息。
%% 类似于Java中的消息拦截器方法。
on_message_publish(Message = #message{topic = Topic, payload = _Payload, from = From}) -> %% 将 Payload 改为 _Payload
    ?SLOG(info, #{msg => "on_message_publish_hook_triggered", topic => Topic, client => From}), %% 添加此行日志
    case select(Message) of
        {true, Querys} when Querys =/= [] -> %% 消息匹配规则，并且有查询目标
            ?SLOG(info, #{msg => "message_matched_rules", topic => Topic, querys => Querys}),
            Context = #{clientid => From},  %% 修改这里：使用clientid作为键名而不是key
            ProducerConf = persistent_term:get(?PRODUCER_CONFIG_KEY, #{persistent_messages => true}), %% 获取存储的生产者配置, 默认true
            PersistentMessages = maps:get(persistent_messages, ProducerConf, true),
            publish_to_rabbitmq(Querys, Context, PersistentMessages);
        SelectResult -> %% 消息不匹配任何规则，或者Querys为空
            ?SLOG(info, #{msg => "message_did_not_match_rules", topic => Topic, select_result => SelectResult}), % 添加日志：消息未匹配规则
            ok %% 不做任何操作
    end,
    {ok, Message}. %% 总是返回 {ok, Message} 给EMQX框架

%% @doc 读取插件的配置文件。
%% 配置文件通常是 HOCON 格式。
read_config() ->
    %% hocon:load/1 解析HOCON配置文件。
    case hocon:load(rabbitmq_config_file()) of
        {ok, RawConf} -> %% 配置文件成功加载并解析
            %% emqx_config:check_config/2 根据预定义的Schema校验配置项的合法性。
            %% emqx_plugin_rabbitmq_schema 是定义配置结构和校验规则的模块。
            case emqx_config:check_config(emqx_plugin_rabbitmq_schema, RawConf) of
                {_, #{plugin_rabbitmq := Conf}} -> %% 配置校验通过，并提取插件相关的配置部分
                    ?SLOG(info, #{ %% 记录加载到的配置信息
                        msg => "emqx_plugin_rabbitmq config",
                        config => Conf
                    }),
                    Conf; %% 返回解析和校验后的配置
                _ -> %% 配置校验失败
                    ?SLOG(error, #{ %% 记录错误
                        msg => "bad_hocon_file",
                        file => rabbitmq_config_file()
                    }),
                    {error, bad_hocon_file}
            end;
        {error, Error} -> %% 配置文件加载或解析失败
            ?SLOG(error, #{ %% 记录错误
                msg => "bad_hocon_file",
                file => rabbitmq_config_file(),
                reason => Error
            }),
            {error, bad_hocon_file}
    end.

%% @doc 获取配置文件的路径。
%% 允许通过环境变量 EMQX_PLUGIN_RabbitMQ_CONF 指定配置文件路径。
%% 否则使用默认路径 "etc/emqx_plugin_rabbitmq.hocon"。
rabbitmq_config_file() ->
    Env = os:getenv("EMQX_PLUGIN_RABBITMQ_CONF"), %% 读取环境变量
    case Env =:= "" orelse Env =:= false of %% 判断环境变量是否为空或未设置
        true -> "etc/emqx_plugin_rabbitmq.hocon"; %% 使用默认路径
        false -> Env %% 使用环境变量指定的路径
    end.

%% @doc 启动插件所需的资源 (RabbitMQ生产者)。
%% Conf: 插件的配置信息。
start_resource(Conf) ->
    ResId = ?PLUGIN_RABBITMQ_RESOURCE_ID, %% 资源ID，在.hrl中定义
    ok = emqx_resource:create_metrics(ResId), %% 为资源创建监控指标
    %% 创建本地资源。emqx_plugin_rabbitmq_producer 是实现资源行为的模块。
    %% ?PLUGIN_RABBITMQ_RESOURCE_GROUP 是资源组ID。
    %% 类似于Java中通过工厂或管理器创建一个服务实例。
    Result = emqx_resource:create_local(
        ResId, %% 资源ID
        ?PLUGIN_RABBITMQ_RESOURCE_GROUP, %% 资源组ID
        emqx_plugin_rabbitmq_producer, %% 实现资源行为的模块
        Conf, %% 传递给资源模块的配置
        #{} %% 额外的选项
    ),
    %% 根据创建结果进行后续处理。
    start_resource_if_enabled(Result).

%% @doc 根据资源创建结果进行后续处理。
%% {ok, _Result = #{error := undefined, id := ResId}}
%% 模式匹配：创建成功，并且结果Map中 'error' 字段为 undefined。
start_resource_if_enabled({ok, _Result = #{error := undefined, id := ResId}}) ->
    {ok, ResId}; %% 返回 {ok, 资源ID}
%% {ok, #{error := Error, id := ResId}}
%% 模式匹配：创建操作本身是ok的，但资源内部初始化可能失败，'error' 字段有值。
start_resource_if_enabled({ok, #{error := Error, id := ResId}}) ->
    ?SLOG(error, #{ %% 记录资源启动错误
        msg => "start resource error",
        error => Error,
        resource_id => ResId
    }),
    emqx_resource:stop(ResId), %% 停止该资源
    error; %% 返回 error 原子，表示启动失败
%% 匹配其他创建失败的情况 (例如 emqx_resource:create_local 直接返回 {error, ...})
start_resource_if_enabled({error, Reason}) ->
    ?SLOG(error, #{msg => "failed_to_create_resource", reason => Reason}),
    {error, Reason}. %% 直接返回错误原因


%% @doc 卸载插件。
%% 当EMQX停止或卸载此插件时调用。
unload() ->
    %% 首先移除之前添加的钩子。
    unhook('message.publish', {?MODULE, on_message_publish}),

    %% 确保ETS表被删除。使用try...catch处理可能发生的错误 (例如表不存在)。
    try
        ets:delete(?PLUGIN_RABBITMQ_TAB)
    catch
        _:_ -> ok %% 忽略删除ETS表时可能发生的任何错误
    end,

    %% 移除本地资源。
    try
        %% 先尝试停止资源。如果资源不存在或已停止，此操作可能返回错误，但我们通常希望继续执行。
        _ = emqx_resource:stop(?PLUGIN_RABBITMQ_RESOURCE_ID),
        %% 然后移除资源定义。
        Result = emqx_resource:remove_local(?PLUGIN_RABBITMQ_RESOURCE_ID),
        %% 清理持久化数据 (如果该插件使用了persistent_term存储状态)。
        persistent_term:erase({?EMQX_PLUGIN_RABBITMQ_APP, ?EMQX_PLUGIN_RABBITMQ_PRODUCERS}),
        Result %% 返回移除结果
    catch
        %% 捕获移除资源过程中可能发生的任何异常。
        Class:Error:Stack ->
            ?SLOG(error, #{msg => "Error removing rabbitmq resource",
                          class => Class,
                          error => Error,
                          stack => Stack}),
            {error, {Class, Error}} %% 返回错误信息
    end.

%% @doc 辅助函数：移除钩子。
%% HookPoint: 钩子点名称。
%% MFA: {Module, Function, Args}，与添加时一致。
unhook(HookPoint, MFA) ->
    emqx_hooks:del(HookPoint, MFA).

%% @doc 解析主题配置规则列表。
%% Topics: 从配置文件中读取的主题规则列表，每个元素是一个Map。

%% 子句1: 如果列表为空，则完成解析。
topic_parse([]) ->
    ok;
%% 子句2: 处理列表中的一个规则。
%% 模式匹配提取规则Map中的字段。
%% Filter: MQTT主题过滤器 (e.g., "sensor/+/data")。
%% RabbitMQTopic: 目标RabbitMQ主题。
%% Tag: RabbitMQ消息标签。
%% RabbitMQMsg: RabbitMQ消息体模板。
%% 修改：根据HOCON配置文件的实际字段进行模式匹配
topic_parse([#{filter := Filter, exchange := Exchange, exchange_type := ExchangeType, routing_key := RoutingKey, rabbitmq_message := RabbitMQMsg} | T]) ->
    %% emqx_placeholder:preproc_tmpl/1 对消息模板进行预处理，可能包含占位符。
    %% 修改：存储所有必要的字段到ETS表
    Item = {Filter, Exchange, ExchangeType, RoutingKey, emqx_placeholder:preproc_tmpl(RabbitMQMsg)},
    %% 将解析后的规则元组插入到ETS表中。
    ets:insert(?PLUGIN_RABBITMQ_TAB, Item),
    %% 递归处理剩余的规则列表 T。
    %% EvtMsg 作为新的 EvtMsgIn 传递下去，供后续匹配的规则复用。
    topic_parse(T);
%% 子句3: 如果列表中的元素不符合期望的Map结构，则跳过该元素，继续处理剩余部分。
%% 这提供了一定的容错性，防止单个错误规则导致整个解析失败。
topic_parse([_ | T]) ->
    topic_parse(T).

%% @doc 根据传入的MQTT消息，从ETS表中选择匹配的规则。
%% Message: #message{} 记录。
select(Message) ->
    %% 从ETS表中获取所有规则。
    Rules = ets:tab2list(?PLUGIN_RABBITMQ_TAB),
    %% 调用select/4辅助函数进行实际的匹配和消息构造。
    %% undefined 表示初始时还没有构造事件消息 (EvtMsg)。
    select(Rules, Message, undefined, []). %% 初始累加器 Acc 为空列表 []

%% @doc 递归处理主题匹配和消息构造 (select的辅助函数)。
%% Rules: 剩余待匹配的规则列表。
%% Message: 当前MQTT消息。
%% EvtMsgIn: 上一轮或初始的事件消息 (用于优化，避免重复构造)。
%% Acc: 累加器，存储已匹配并构造好的 {RabbitMQTopic, Tag, Msg} 元组列表。

%% 子句1: 如果规则列表为空，表示所有规则都已处理完毕。
select([], _, _, Acc) ->
    {true, Acc}; %% 返回true和累积的结果列表

%% 子句2: 处理当前规则。
%% {Filter, RabbitMQTopic, Tag, RabbitMQMsgTmpl} 从规则元组中提取信息。
%% 修改：根据ETS表中存储的字段进行模式匹配
select([{Filter, Exchange, ExchangeType, RoutingKey, RabbitMQMsgTmpl} | T], Message, EvtMsgIn, Acc) ->
    %% match_topic/2 判断当前MQTT消息的主题是否与规则中的Filter匹配。
    case match_topic(Message, Filter) of
        true -> %% 主题匹配成功
            %% 优化：仅在首次匹配成功时构造 EvtMsg。
            %% 如果 EvtMsgIn 已经是 undefined，说明这是第一个匹配到的规则，
            %% 则调用 ?evt_mod:eventmsg_publish(Message) 来创建包含消息详情的 EvtMsg。
            %% 否则，复用之前创建的 EvtMsgIn。
            EvtMsg = case EvtMsgIn of
                         undefined -> ?evt_mod:eventmsg_publish(Message);
                         _ -> EvtMsgIn
                     end,
            %% emqx_plugin_rabbitmq_util:render/2 使用 EvtMsg 中的信息渲染 rabbitmq 消息体模板。
            Payload = emqx_plugin_rabbitmq_util:render(RabbitMQMsgTmpl, EvtMsg),
            %% 构建发送给生产者的消息属性Map
            MessageProps = #{
                exchange => Exchange,
                routing_key => RoutingKey,
                exchange_type => ExchangeType
                %% persistent_messages 从生产者配置中获取，不需要在这里传递
            },
            %% 递归处理剩余规则 T，将当前构造好的 {MessageProps, Payload} 加入累加器 Acc。
            %% EvtMsg 作为新的 EvtMsgIn 传递下去，供后续匹配的规则复用。
            select(T, Message, EvtMsg, [{MessageProps, Payload} | Acc]);
        false -> %% 主题不匹配
            %% 递归处理剩余规则 T，累加器 Acc 和 EvtMsgIn 保持不变。
            select(T, Message, EvtMsgIn, Acc)
    end.

%% @doc 主题匹配逻辑。
%% 第一个参数通常是 #message{} 记录，第二个参数是主题过滤器字符串。

%% 子句1 & 2: 过滤掉以 "$#" 或 "$+" 开头的过滤器 (这些通常是MQTT保留的，不用于常规匹配)。
match_topic(_, <<"$#", _/binary>>) -> false;
match_topic(_, <<"$+", _/binary>>) -> false;
%% 子句3: 如果消息主题是系统主题 (以 "$SYS/" 开头)，则不匹配任何用户定义的过滤器。
match_topic(#message{topic = <<"$SYS/", _/binary>>}, _) -> false;
%% 子句4: 正常的MQTT主题与过滤器匹配。
%% #message{topic = Topic} 从消息记录中提取主题。
%% emqx_topic:match/2 是EMQX提供的标准主题匹配函数，支持MQTT通配符 '+' 和 '#'。
match_topic(#message{topic = Topic}, Filter) ->
    emqx_topic:match(Topic, Filter);
%% 子句5: 其他不符合上述情况的，均视为不匹配。
match_topic(_, _) -> false.

%% @doc 执行查询，即将消息发送到RabbitMQ。
%% Querys: 一个列表，每个元素是 {MessageProps, Payload}。
%% Context: 包含额外信息的Map，如消息来源客户端ID。
%% PersistentMessages: 从配置中读取的消息持久化标志。
publish_to_rabbitmq([], _Context, _PersistentMessages) ->
    ok;
publish_to_rabbitmq([{MessageProps, ActualPayload} | RestQuerys], Context, PersistentMessages) ->
    Exchange = maps:get(exchange, MessageProps),
    RoutingKey = maps:get(routing_key, MessageProps),
    %% ExchangeType = maps:get(exchange_type, MessageProps), %% ExchangeType 由生产者侧处理或配置

    %% 不需要修改Context，直接传递
    SingleQuery = {publish, Exchange, RoutingKey, PersistentMessages, ActualPayload, Context},
    ?SLOG(debug, #{msg => "sending_single_query_to_rabbitmq", query => SingleQuery}),
    Result = emqx_resource:query(?PLUGIN_RABBITMQ_RESOURCE_ID, SingleQuery, #{}),
    query_ret(Result, SingleQuery, Context),
    publish_to_rabbitmq(RestQuerys, Context, PersistentMessages).

%% @doc 处理查询结果。
%% Result: emqx_resource:query/3 的返回结果。
%% OriginalQuery: 发送给资源的原始单个查询。
%% Context: 上下文信息。
query_ret({ok, _}, _OriginalQuery, _Context) ->
    %% 成功处理，可以添加日志或指标记录
    ?SLOG(debug, #{msg => "rabbitmq_resource_query_successful", query => _OriginalQuery}),
    ok;
query_ret({async_return, _}, _OriginalQuery, _Context) ->
    %% 处理异步返回结果
    ?SLOG(debug, #{msg => "rabbitmq_resource_query_async_successful", query => _OriginalQuery}),
    ok;
query_ret({error, Reason}, OriginalQuery, Context) ->
    %% 处理错误情况，记录日志
    ?SLOG(error, #{msg => "rabbitmq_resource_query_failed", reason => Reason, query => OriginalQuery, context => Context}),
    {error, Reason}.